package com.yh.csx.bsf.ods;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.yh.csx.bsf.core.thread.ThreadPool;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.monitor.CheckSchedule;
import com.yh.csx.bsf.ods.utils.WarnUtils;
import com.yh.csx.bsf.redis.RedisProvider;

import lombok.Getter;
import lombok.Setter;

/**
 * 增量同步
 */

public class CanalIncrementConsumer extends AbstractCanalConsumer {

	
	@Getter
	@Setter
	Map<String, OdsSyncAdapter> syncers;
	@Getter
	@Setter
	com.yh.csx.bsf.redis.RedisProvider redisProvider;
	@Getter
	@Setter
	StragtegyProvider stragtegyProvider;

	/**
	 * 创建表
	 * 
	 * @param header 可以从header中获得schema、table的名称
	 * @param sql
	 */
	public void createTable(CanalEntry.Header header, String sql) {	
	}

	/**
	 * 修改表结构,即alter指令，需要声明：通过alter增加索引、删除索引，也是此操作。
	 * 
	 * @param header 可以从header中获得schema、table的名称
	 * @param sql
	 */
	public void alterTable(CanalEntry.Header header, String sql) {	
	}

	/**
	 * 清空、重建表
	 * 
	 * @param header 可以从header中获得schema、table的名称
	 * @param sql
	 */
	public void truncateTable(CanalEntry.Header header, String sql) {
	}
	
	public CanalIncrementConsumer(Map<String, OdsSyncAdapter> syncers, RedisProvider redisProvider,
			StragtegyProvider stragtegyProvider) {
		super();
		this.syncers = Optional.ofNullable(syncers).orElseThrow(RuntimeException::new);// 如果为抛异常
		this.redisProvider = redisProvider;
		this.stragtegyProvider = stragtegyProvider;
	}

	@Override
	public void insert(Header header, List<Column> afterColumns) throws Exception {
		LogUtils.debug(CanalIncrementConsumer.class, "ods", "insert event");
		if (redisProvider != null) {
			if(redisProvider.exists(stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()))) {
				//Thread.sleep(2000);
				throw new Exception("redis 锁存在"+stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()));
			}
		}
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().insert(header, afterColumns);
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" + stragtegyProvider.getTable(header.getSchemaName() , header.getTableName()) + "结束", e);
				//throw e;
				String id="";
				for(Column col:afterColumns) {
					if(col.getName().equalsIgnoreCase("id")) {
						id=col.getValue();
					}
				}
				WarnUtils.alarm("增量同步异常",header.getSchemaName()+"."+header.getTableName()+":insert:"+id , e);
			}
		}

	}

	@Override
	public void update(Header header, List<Column> beforeColumns, List<Column> afterColumns) throws Exception {
		LogUtils.debug(CanalIncrementConsumer.class, "ods", "update event");
		if (redisProvider != null) {
			if(redisProvider.exists(stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()))) {
				//Thread.sleep(2000);
				throw new Exception("redis 锁存在"+stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()));
			}
		}
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().update(header, beforeColumns, afterColumns);
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" + stragtegyProvider.getTable(header.getSchemaName() , header.getTableName()) + "结束", e);
				//throw e;
				String id="";
				for(Column col:afterColumns) {
					if(col.getName().equalsIgnoreCase("id")) {
						id=col.getValue();
					}
				}
				WarnUtils.alarm("增量同步异常",header.getSchemaName()+"."+header.getTableName()+":update:"+id , e);
				throw e;
			}
		}
	}

	@Override
	public void delete(Header header, List<Column> beforeColumns) throws Exception {
		LogUtils.debug(CanalIncrementConsumer.class, "ods", "delete event");
		if (redisProvider != null) {
			
			if(redisProvider.exists(stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()))) {
				throw new Exception("redis 锁存在"+stragtegyProvider.getTable(header.getSchemaName() , header.getTableName()));
				//Thread.sleep(2000);
			}
		}
		for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
			try {
				syncer.getValue().delete(header, beforeColumns);
			} catch (Exception e) {
				LogUtils.error(this.getClass(), "dds", "【增量同步】异常" + stragtegyProvider.getTargetTable(header.getSchemaName() , header.getTableName()) + "结束", e);
				//throw e;
				String id="";
				for(Column col:beforeColumns) {
					if(col.getName().equalsIgnoreCase("id")) {
						id=col.getValue();
					}
				}
				WarnUtils.alarm("增量同步异常",header.getSchemaName()+"."+header.getTableName()+":insert:"+id , e);
				throw e;
			}
		}

	}

	@Override
	public void connectorBatchArk(long batchId) throws Exception {

	}

	@Override
	public boolean insertBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);
		if(CheckSchedule.CHECK_TABLE_TASKS.containsKey(targetTable)) {
			LogUtils.debug(CanalIncrementConsumer.class, "ods", "update event");
			if (redisProvider != null) {
				if(redisProvider.exists(targetTable)) {
					//Thread.sleep(2000);
					throw new Exception("redis 锁存在"+targetTable);
				}
			}
			for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
				try {
					syncer.getValue().insertBatch(targetTable,entries); 
				} catch (Exception e) {
					LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
					WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);					
					throw e;				
				}
			}
		}
		return true;
	}

	@Override
	public boolean updateBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);
		if(CheckSchedule.CHECK_TABLE_TASKS.containsKey(targetTable)) {
			LogUtils.debug(CanalIncrementConsumer.class, "ods", "update event");
			if (redisProvider != null) {
				if(redisProvider.exists(targetTable)) {
					//Thread.sleep(2000);
					throw new Exception("redis 锁存在"+targetTable);
				}
			}
			for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
				try {
					syncer.getValue().updateBatch(targetTable,entries); 
				} catch (Exception e) {
					LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
					WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);					
					throw e;
				}
			}
		}
		return true;
	}

	@Override
	public boolean deleteBatch(com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries) throws Exception {
		String database=entries[0].getHeader().getSchemaName();
		String table=entries[0].getHeader().getTableName();
		String targetTable=stragtegyProvider.getTargetTable(database, table);
		if(CheckSchedule.CHECK_TABLE_TASKS.containsKey(targetTable)) {
			LogUtils.debug(CanalIncrementConsumer.class, "ods", "update event");
			if (redisProvider != null) {
				if(redisProvider.exists(targetTable)) {
					//Thread.sleep(2000);
					throw new Exception("redis 锁存在"+targetTable);
				}
			}
			for (Entry<String, OdsSyncAdapter> syncer : syncers.entrySet()) {
				try {
					syncer.getValue().deleteBatch(targetTable,entries); 
				} catch (Exception e) {
					LogUtils.error(this.getClass(), "dds", "【增量同步】异常" +targetTable + "", e);
					WarnUtils.alarm("增量同步异常",database+"."+table+":update:", e);
					try {Thread.sleep(1000);}catch(Exception ex) {}
					throw e;
				}
			}
		}
		return true;
	}
	
	@Override
	public void whenOthers(CanalEntry.Entry entry) {
		try {
			long delay = System.currentTimeMillis() - entry.getHeader().getExecuteTime();
			LogUtils.debug(this.getClass(), "dds", "【增量同步】" + entry.getHeader().getSchemaName() + "延迟" + delay + "ms");
		}catch(Exception e) {
			
		}
		
	}
	@Override
	protected boolean processMessages(Message message) throws Exception {
		Map<String, List<CanalEntry.Entry>> mappingList = new HashMap<String, List<CanalEntry.Entry>>();
		for (CanalEntry.Entry entry : message.getEntries()) {
			if (entry.getEntryType().equals(EntryType.ROWDATA)) {
				String targetTable=stragtegyProvider.getTargetTable(entry.getHeader().getSchemaName() , entry.getHeader().getTableName());
				if (mappingList.containsKey(targetTable)) {
					mappingList.get(targetTable)
							.add(entry);
				} else {
					List<CanalEntry.Entry> list = new ArrayList<CanalEntry.Entry>();
					list.add(entry);
					mappingList.put(targetTable, list);
				}
			} else {
				whenOthers(entry);
			}
		}
		return processMessages0(mappingList);

	}	
}
